package com.binarysoft.cmdhandler.common.hbase;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.StringTokenizer;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellScanner;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
import org.apache.hadoop.hbase.util.Bytes;

import com.binarysoft.service.common.hbase.HBaseService;
import common.framework.command.handler.AbstractCommandHandler;
import common.framework.command.helper.CommandHelper;
import common.framework.command.server.Terminal;
import common.framework.dsb.annotation.DSBCmd;
import common.framework.dsb.client.DefaultServiceFactory;
import common.framework.dsb.client.ServiceFactory;

@DSBCmd("hbase")
public class HBaseCmdHandler extends AbstractCommandHandler {

	@Override
	public void handle(String args, Terminal terminal) throws Exception {
		terminal.println("");
		StringTokenizer stier = new StringTokenizer(args);
		int count = stier.countTokens();
		if (count < 2) {
			terminal.println(" Usage:");
			terminal.println(" hbase cluster                                               -- list all HBase clusters.");
			terminal.println(" hbase scan        <cluster_name> <table_name>               -- scan whole hbase table.");
			terminal.println(" hbase scantest    <cluster_name> <table_name>               -- scan test performance.");
			terminal.println(" hbase get         <cluster_name> <table_name>               -- get data from table.");
			terminal.println(" hbase put         <cluster_name> <table_name>               -- put data into table.");
			terminal.println(" hbase delete      <cluster_name> <table_name>               -- delete data from table.");
			terminal.println(" hbase list        <cluster_name>                            -- list all tables.");
			terminal.println(" hbase desc        <cluster_name> <table_name>               -- list all tables.");
			terminal.println(" hbase dummy       <cluster_name> <table_name>               -- dummy rows.");
			terminal.println("");
			return;
		}
		stier.nextToken();
		String cmd = stier.nextToken();
		if ("cluster".equalsIgnoreCase(cmd) && count == 2) {

			ServiceFactory serviceFactory = new DefaultServiceFactory();
			HBaseService hbaseService = serviceFactory.lookup("DSB/HBaseService");

			String[] tableHAD = { "Cluster Name" };
			String[] tableGAP = { "------------" };
			int[] format = { 40 };
			terminal.println(CommandHelper.format(tableHAD, format));
			terminal.println(CommandHelper.format(tableGAP, format));

			Collection<String> clusters = hbaseService.listClusters();

			for (String cluster : clusters) {
				Object[] row = { cluster };
				terminal.println(CommandHelper.format(row, format));
			}

		} else if ("get".equalsIgnoreCase(cmd) && count == 4) {

			ServiceFactory serviceFactory = new DefaultServiceFactory();
			HBaseService hbaseService = serviceFactory.lookup("DSB/HBaseService");

			String clusterName = stier.nextToken();
			String tableName = stier.nextToken();
			Table table = hbaseService.getHtable(clusterName, tableName);

			if (table == null) {
				terminal.print("Table [" + tableName + "] was not found!");
				terminal.println("");
				return;
			}

			terminal.print("input row key(*):");
			String rowKey = terminal.read();

			if (isEmpty(rowKey)) {
				terminal.println("Row key is mandatory field!");
				terminal.println("");
				return;
			}

			terminal.print("input column family:");
			String familyName = terminal.read();

			terminal.print("input column name:");
			String columnName = terminal.read();

			Get get = new Get(Bytes.toBytes(rowKey));

			if (!isEmpty(familyName)) {
				// 获取指定列族
				get.addFamily(Bytes.toBytes(familyName));
			}
			if (!isEmpty(familyName) && !isEmpty(columnName)) {
				// 获取指定列族和列修饰符对应的列
				get.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columnName));
			}
			terminal.println("");
			getTable(table, get, terminal);

		} else if ("put".equalsIgnoreCase(cmd) && count == 4) {

			ServiceFactory serviceFactory = new DefaultServiceFactory();
			HBaseService hbaseService = serviceFactory.lookup("DSB/HBaseService");

			String clusterName = stier.nextToken();
			String tableName = stier.nextToken();
			Table table = hbaseService.getHtable(clusterName, tableName);

			if (table == null) {
				terminal.print("Table [" + tableName + "] was not found!");
				terminal.println("");
				return;
			}

			terminal.print("input row key(*):");
			String rowKey = terminal.read();

			if (isEmpty(rowKey)) {
				terminal.println("Row key is mandatory field!");
				terminal.println("");
				return;
			}

			terminal.print("input column family:");
			String familyName = terminal.read();

			terminal.print("input column name:");
			String columnName = terminal.read();

			terminal.print("input value:");
			String value = terminal.read();

			Put put = new Put(Bytes.toBytes(rowKey));
			put.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columnName), Bytes.toBytes(value));
			put.setDurability(Durability.SYNC_WAL);
			table.put(put);
			table.close();

			terminal.println("Put data into table [" + tableName + "] success.");

		} else if ("delete".equalsIgnoreCase(cmd) && count == 4) {

			ServiceFactory serviceFactory = new DefaultServiceFactory();
			HBaseService hbaseService = serviceFactory.lookup("DSB/HBaseService");

			String clusterName = stier.nextToken();
			String tableName = stier.nextToken();
			Table table = hbaseService.getHtable(clusterName, tableName);

			if (table == null) {
				terminal.print("Table [" + tableName + "] was not found!");
				terminal.println("");
				return;
			}

			terminal.print("input row key(*):");
			String rowKey = terminal.read();

			if (isEmpty(rowKey)) {
				terminal.println("Row key is mandatory field!");
				terminal.println("");
				return;
			}

			terminal.print("input column family:");
			String familyName = terminal.read();

			terminal.print("input column name:");
			String columnName = terminal.read();

			Delete delete = new Delete(Bytes.toBytes(rowKey));

			if (!isEmpty(familyName)) {
				delete.addFamily(Bytes.toBytes(familyName));
			}

			if (!isEmpty(familyName) && !isEmpty(columnName)) {
				delete.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columnName));
			}

			table.delete(delete);
			table.close();

			terminal.println("Delete data from table [" + tableName + "] success.");

		} else if ("scan".equalsIgnoreCase(cmd) && count == 4) {
			ServiceFactory serviceFactory = new DefaultServiceFactory();
			HBaseService hbaseService = serviceFactory.lookup("DSB/HBaseService");

			String clusterName = stier.nextToken();
			String tableName = stier.nextToken();
			Table table = hbaseService.getHtable(clusterName, tableName);

			if (table == null) {
				terminal.print("Table [" + tableName + "] was not found!");
				terminal.println("");
				return;
			}
			scanTable(table, terminal);
		} else if ("scantest".equalsIgnoreCase(cmd) && count == 4) {
			ServiceFactory serviceFactory = new DefaultServiceFactory();
			HBaseService hbaseService = serviceFactory.lookup("DSB/HBaseService");

			String clusterName = stier.nextToken();
			String tableName = stier.nextToken();
			Table table = hbaseService.getHtable(clusterName, tableName);

			if (table == null) {
				terminal.print("Table [" + tableName + "] was not found!");
				terminal.println("");
				return;
			}
			scanTableTest(table, terminal);
		} else if ("list".equalsIgnoreCase(cmd) && count == 3) {
			ServiceFactory serviceFactory = new DefaultServiceFactory();
			HBaseService hbaseService = serviceFactory.lookup("DSB/HBaseService");
			String clusterName = stier.nextToken();
			Collection<String> names = hbaseService.listTableNames(clusterName);
			int i = 0;
			for (String tableName : names) {
				i++;
				terminal.println(tableName);
			}
			terminal.println("");
			terminal.println(" " + i + " table(s) was found.");
		} else if ("desc".equalsIgnoreCase(cmd) && count == 4) {
			ServiceFactory serviceFactory = new DefaultServiceFactory();
			HBaseService hbaseService = serviceFactory.lookup("DSB/HBaseService");
			String clusterName = stier.nextToken();
			String tableName = stier.nextToken();

			HTableDescriptor htdesc = hbaseService.getTableDescriptor(clusterName, tableName);
			HColumnDescriptor[] hcdescs = htdesc.getColumnFamilies();
			StringBuffer sbuffer = new StringBuffer();
			sbuffer.append("Row key");
			for (HColumnDescriptor hcdesc : hcdescs) {
				sbuffer.append(", " + hcdesc.getNameAsString());
			}
			terminal.println("Table name: " + tableName);
			terminal.println(sbuffer.toString());
		} else if ("dummy".equalsIgnoreCase(cmd) && count == 4) {

			ServiceFactory serviceFactory = new DefaultServiceFactory();
			HBaseService hbaseService = serviceFactory.lookup("DSB/HBaseService");

			String clusterName = stier.nextToken();
			String tableName = stier.nextToken();

			String[] families = new String[] { "cpu", "memory", "disk" };

			java.util.Random random = new java.util.Random();// 定义随机类

			terminal.print("Please input number of dummy datas: ");
			int dataCount = Integer.parseInt(terminal.read());

			long start = System.currentTimeMillis();

			terminal.println("Dummy data begin at : " + CommandHelper.getTime(start));
			Table table = hbaseService.getHtable(clusterName, tableName);
			List<Put> puts = new ArrayList<Put>();

			for (int i = 1; i <= dataCount; i++) {
				String rowKey = String.valueOf(System.nanoTime());
				int familyIndex = random.nextInt(3);// 返回[0,3)集合中的整数，注意不包括3
				String familyName = families[familyIndex];
				long id = System.nanoTime();
				String columnName = String.valueOf(id);
				String value = String.valueOf(random.nextFloat() + 10000);

				Put put = new Put(Bytes.toBytes(rowKey));
				put.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columnName), Bytes.toBytes(value));
				put.setDurability(Durability.SYNC_WAL);
				puts.add(put);
			}
			table.put(puts);
			table.close();
			long end = System.currentTimeMillis();
			terminal.println("Dummy data counts : " + dataCount);
			terminal.println("Dummy data finished at : " + CommandHelper.getTime(end));
			terminal.println("Dummy data consume: " + (end - start) + " mili-seconds.");
		}

		terminal.println("");
	}

	private void getTable(Table table, Get get, Terminal terminal) throws Exception {
		Result result = table.get(get);

		if (result.isEmpty()) {
			terminal.print("0 row(s) was found.");
			terminal.println("");
			return;
		}

		String rowKey = Bytes.toString(result.getRow());

		String[] tableHAD = { "Row Key", "Family:Column", "Timestamp", "Value" };
		String[] tableGAP = { "-------", "-------------", "---------", "-----" };
		int[] format = { 70, 35, 35, 35 };
		terminal.println(CommandHelper.format(tableHAD, format));
		terminal.println(CommandHelper.format(tableGAP, format));

		CellScanner scaner = result.cellScanner();
		while (scaner.advance()) {
			Cell cell = scaner.current();
			String family = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
			String qualifier = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
			String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
			long timeStamp = cell.getTimestamp();
			Object[] output = new Object[] { rowKey, family + ":" + qualifier, CommandHelper.getTime(timeStamp), value };
			terminal.println(CommandHelper.format(output, format));
		}
		table.close();
	}

	private void scanTable(Table table, Terminal terminal) throws Exception {
		String[] tableHAD = { "Row Key", "Family:Column", "Timestamp", "Value" };
		String[] tableGAP = { "-------", "-------------", "---------", "-----" };
		int[] format = { 70, 35, 35, 35 };
		terminal.println(CommandHelper.format(tableHAD, format));
		terminal.println(CommandHelper.format(tableGAP, format));

		int rowCount = 0;
		Scan scan = new Scan();
		scan.setBatch(50); // max 50 rows
		scan.setFilter(new FirstKeyOnlyFilter());

		long start = System.currentTimeMillis();
		terminal.println("scan data begin at : " + CommandHelper.getTime(start));

		ResultScanner resultScanner = table.getScanner(scan);
		try {
			for (Result result : resultScanner) {
				rowCount += result.size();
				// terminal.println("Row->: " + rowCount);
				String rowKey = Bytes.toString(result.getRow());
				CellScanner scaner = result.cellScanner();
				while (scaner.advance()) {
					Cell cell = scaner.current();
					String family = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
					String qualifier = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
					String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
					long timeStamp = cell.getTimestamp();
					Object[] output = new Object[] { rowKey, family + ":" + qualifier, CommandHelper.getTime(timeStamp), value };
					// terminal.println(CommandHelper.format(output, format));
				}
			}
		} finally {
			resultScanner.close();
			table.close();
		}

		long end = System.currentTimeMillis();
		terminal.println("Scan row counts: : " + rowCount + " rows.");
		terminal.println("Scan data finished at : " + CommandHelper.getTime(end));
		terminal.println("Scan consume: " + (end - start) + " mili-seconds.");
	}

	private void scanTableTest(Table table, Terminal terminal) throws Exception {

		int rowCount = 0;
		Scan scan = new Scan();
		scan.setBatch(50); // max 50 rows
		long start = System.currentTimeMillis();
		long end = 0;

		terminal.println(" Fetch row count begin at : " + CommandHelper.getTime(start));
		scan.setFilter(new FirstKeyOnlyFilter());
		ResultScanner resultScanner = table.getScanner(scan);
		try {
			for (Result result : resultScanner) {
				rowCount += result.size();
			}

		} finally {
			resultScanner.close();
			table.close();
		}
		end = System.currentTimeMillis();
		terminal.println(" Row count: " + rowCount + " rows.");
		terminal.println(" Fetch row count finished at : " + CommandHelper.getTime(end));
		terminal.println(" Fetch row count consume: " + (end - start) + " mili-seconds.");
		terminal.println("---------------------------------->");

		String[] tableHAD = { " Read Rows", "MAX time", "MIN time", "AVG time" };
		String[] tableGAP = { " ----------", "--------", "--------", "--------" };
		int[] format = { 25, 25, 25, 25 };
		terminal.println(CommandHelper.format(tableHAD, format));
		terminal.println(CommandHelper.format(tableGAP, format));

		int countStep = 100;
		long[] testResult = new long[10];
		for (int i = countStep; i <= 1000000; i = i * 10) {

			for (int j = 0; j < 10; j++) {
				long time = scanTest(table, terminal, i);
				testResult[j] = time;
			}

			Object[] iResult = { i, getMax(testResult), getMin(testResult), getAvg(testResult) };
			terminal.println(CommandHelper.format(iResult, format));
		}
	}

	private long scanTest(Table table, Terminal terminal, int targetRows) throws Exception {
		Scan scan = new Scan();
		scan.setBatch(50); // max 50 rows
		scan.setFilter(new FirstKeyOnlyFilter());
		
		long start = System.currentTimeMillis();
		ResultScanner rs = table.getScanner(scan);
		try {
			int readRowCount = 0;
			for (Result result : rs) {
				readRowCount += result.size();
				if (readRowCount >= targetRows) {
					break;
				}
			}
		} finally {
			rs.close();
			table.close();
		}
		long end = System.currentTimeMillis();
		return end - start;
	}

	private boolean isEmpty(String s) throws Exception {
		return s == null || "".equalsIgnoreCase(s.trim());
	}

	private long getMax(long[] data) {
		long max = data[0];
		for (int i = 0; i < data.length; i++) {
			if (data[i] > max) {
				max = data[i];
			}
		}
		return max;
	}

	private long getMin(long[] data) {
		long min = data[0];
		for (int i = 0; i < data.length; i++) {
			if (data[i] < min) {
				min = data[i];
			}
		}
		return min;
	}

	private long getAvg(long[] data) {
		long sum = 0;
		for (int i = 0; i < data.length; i++) {
			sum += data[i];
		}
		return sum / data.length;
	}
}
